/*
 * $AIST_Release: 0.9.0 $
 * Copyright 2011 Information Technology Research Institute, National
 * Institute of Advanced Industrial Science and Technology
 * 
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 * 
 *    http://www.apache.org/licenses/LICENSE-2.0
 * 
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package aggr

import java.io.PrintWriter
import java.io.FileInputStream
import java.io.FileOutputStream
import java.io.IOException
import java.lang.Boolean.parseBoolean

import scala.collection.JavaConversions._

import com.google.protobuf.Descriptors._
import com.google.protobuf.DescriptorProtos._
import com.google.protobuf.CodedOutputStream

import aggr.Aggregators._
import aggr.{SerializedTypes => ST}
import pbdata.Tables
import util.PBUtils._
import util.KeyValueList

/**
 * Module for entry point of aggregator process.
 */
object AggrMain {

  /**
   * Entry point of aggregator process.
   */
  def main(args: Array[String]) {
    args match {
      case Array(x, y, z, _*) => aggregate(x, y, z)
      case _ => 
        System.err.println("no format file or data file.")
        System.exit(1)
    }
  }

  /**
   * Do aggregates.
   *
   * @param outputFile The file to write results.
   * @param formatFile The file where input data format has been described.
   * @param inputFile  The file where input data has been described.
   */
  def aggregate(outputFile: String, formatFile: String, inputFile: String): Unit = {

    val format = Tables.DataFormat.parseFrom(new FileInputStream(formatFile))

    val protoFiles = try {
      convertFiles(FileDescriptorSet.parseFrom(format.getProtoFiles())).values().toList
    } catch {
      case e: PBConvertingException => 
        System.err.println(e)
        exit(1)
      case e: com.google.protobuf.Descriptors.DescriptorValidationException => 
        System.err.println(e)
        exit(1)
    }

    val typeMap = ST.makeSTypeMap(protoFiles)

    val keyType   = typeMap(format.getKeyType)
    val valType   = typeMap(format.getValueType)
    val tableArgs = format.getTableArgsList().toList.map(_.longValue)

    val kv = new KeyValueList(new FileInputStream(inputFile)).map { kv =>
      // lazy evaluation
      (Deserializer.deserialize(keyType, kv.getKey), 
       Deserializer.deserialize(valType, kv.getValue)) 
    }

    val className = System.getProperty("sc.aggr")
    if (className == null) {
        System.err.println("sc.aggr property is empty.")
        System.exit(1)
    }
    val factory = Class.forName(className).newInstance().asInstanceOf[AggregatorFactory]
    val aggr = factory.createNew(keyType, valType, tableArgs)

    val outputText = {
      val v = System.getProperty("sc.outputText")
      if (v != null) parseBoolean(v)
      else           false
    }

    val rslt = doPrint(outputFile, 
      aggr.getKeyType.getOrElse(keyType),
      aggr.getValueType.getOrElse(valType),
      outputText) { writer =>

      aggr.aggregate(kv)(writer)
    }

    if (!rslt) {
      exit(1)
    }
  }

  /**
   * Open file for writing.
   */
  def doPrint(
    outputFile:String, 
    keyType: ST.SerializedType, 
    valType: ST.SerializedType, 
    outputText: Boolean)(f: ((Any, Any) => Unit) => Unit): Boolean = {

    try {
      if (outputText) {
        val os = new PrintWriter(outputFile)
        try {
          f { (key, value) => 
            os.println("key: %s".format(key))
            os.println("value: %s".format(value))
          }
        } finally {
          os.close()
        }
      } else {
        val os = new FileOutputStream(outputFile)
        try {
          val cos = CodedOutputStream.newInstance(os)

          f { (key, value) => 
            cos.writeMessage(1, 
              Tables.KeyValue.newBuilder().
                setKey(Serializer.serialize(keyType, key)).
                setValue(Serializer.serialize(valType, value)).build())
          }
          cos.flush()
        } finally {
          os.close()
        }
      }
      println("Completes writing to \"%s\".".format(outputFile))
      true
    } catch {
      case e: IOException => 
        print(e)
        false
    }
  }
}
